package com.atguigu.flink.edu.app.fun;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.flink.edu.common.GmallConfig;
import com.atguigu.flink.edu.utils.DimUtil;
import com.atguigu.flink.edu.utils.PhoenixUtil;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.util.Collection;
import java.util.Set;

public class DimSinkFunction implements SinkFunction<JSONObject> {
    @Override
    public void invoke(JSONObject jsonObject, Context out) throws Exception {
        String sink_table = jsonObject.getString("sink_table");
        jsonObject.remove("sink_table");
     String type = jsonObject.getString("type");
     jsonObject.remove("type");
        Set<String> keySet = jsonObject.keySet();
        Collection<Object> values = jsonObject.values();
        String insertSql = " upsert into "+ GmallConfig.PHOENIX_SCHEMA + "." +
                sink_table + "(" + StringUtils.join(keySet,",")
                + ")" + " values ('" + StringUtils.join(values,"','") + "')";
        System.out.println("像Phoenix中插入的sql"+ insertSql);
        PhoenixUtil.excutSql(insertSql);
        //if (type.equals("update")){
        //    DimUtil.deleteCached(sink_table,jsonObject.getString("id"));
        //}
    }
}